package com.shujia.flink.tf;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;

public class Demo05Reduce {
    public static void main(String[] args) throws Exception {
        /*
         * reduce算子：对KeyBy之后的数据流按照Key对Value进行聚合
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> socketDS = env.socketTextStream("master", 8888);

        // 统计单词数量
        socketDS.map(w -> Tuple2.of(w, 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(kv -> kv.f0, Types.STRING)
                // 自定义聚合方法：可以实现sum、max、min，也可以直接KeyBy之后调用sum、max、min方法完成计算
//                .reduce((t1, t2) -> Tuple2.of(t1.f0, t1.f1 + t2.f1)).print();
//                .sum(1).print();
//                .max(1).print();
                .min(1).print();

        // 统计每个班级学生的平均年龄
        ArrayList<String> arr = new ArrayList<>();
        arr.add("文科一班,20");
        arr.add("文科一班,21");
        arr.add("文科一班,22");
        arr.add("文科二班,24");
        arr.add("文科二班,22");
        arr.add("文科二班,23");

        DataStreamSource<String> stuDS = env.fromCollection(arr);

        stuDS.map(s -> Tuple3.of(s.split(",")[0], Integer.parseInt(s.split(",")[1]), 1), Types.TUPLE(Types.STRING, Types.INT, Types.INT))
                .keyBy(t3 -> t3.f0, Types.STRING)
                .reduce((t1, t2) -> Tuple3.of(t1.f0, t1.f1 + t2.f1, t1.f2 + t2.f2))
                .map(t3 -> Tuple2.of(t3.f0, t3.f1 * 1.0 / t3.f2), Types.TUPLE(Types.STRING, Types.DOUBLE))
                .print();

        env.execute();
    }
}
